package com.mojo.kafka.configuration;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.*;
import org.springframework.util.backoff.BackOff;
import org.springframework.util.backoff.FixedBackOff;

import java.util.HashMap;
import java.util.Map;

/**
 * @author <a href="mailto:mojo_jj@163.com">Mojo</a>
 */
@Configuration
public class KafkaConfiguration {
    private static final Logger logger = LoggerFactory.getLogger(KafkaConfiguration.class);

    @Bean
    @Primary
    public ErrorHandler kafkaErrorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
        ConsumerRecordRecoverer recover = new DeadLetterPublishingRecoverer(kafkaTemplate);
        // 设置重试间隔 10秒 次数为 3次
        BackOff backOff = new FixedBackOff(10 * 1000L, 1L);
        // 创建 SeekToCurrentErrorHandler 对象
        return new SeekToCurrentErrorHandler(recover, backOff);
    }

    @Bean("batchConsumerFactory")
    public KafkaListenerContainerFactory factory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
        // 开启批量消费
        factory.setBatchListener(true);
        // 关闭自启
        factory.setAutoStartup(true);
        factory.setBatchErrorHandler((ex, records) -> {
            for (ConsumerRecord<?, ?> record : records) {
                logger.error("消费失败:{}", record.value());
            }
        });
        return factory;
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> consumerConfig = new HashMap<>();

        consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "47.106.130.181:9092");
        consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "defaultConsumerGroup");
        consumerConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 12000);
        consumerConfig.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 18000);
        consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        //consumerConfig.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, this.maxPartitionFetchBytes);

        return consumerConfig;
    }
}
